package cn.dmrliu.edu.realtime.app.dws;

import cn.dmrliu.edu.realtime.app.func.DimAsyncFunction;
import cn.dmrliu.edu.realtime.bean.VideoPlayBean;
import cn.dmrliu.edu.realtime.util.ClickhouseUtil;
import cn.dmrliu.edu.realtime.util.DateUtil;
import cn.dmrliu.edu.realtime.util.KafkaUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * 要使用会话窗口
 */
public class DwsToolVideoWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1. 基本环境的准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);


        // TODO 2. 检查点的设置

        // TODO 3. 从kafka 观看视频主题读取数据
        String topic = "edu_dwd_tool_video";
        String group = "dws_tool_video_window_group";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, group);
        DataStreamSource<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka souerce");

        // TODO 4. 转换为实体类，补充维度信息
        SingleOutputStreamOperator<VideoPlayBean> beanDS = kafkaDS.map(new MapFunction<String, VideoPlayBean>() {
            @Override
            public VideoPlayBean map(String value) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                String videoId = jsonObject.getJSONObject("appVideo").getString("video_id");
                Long playSec = jsonObject.getJSONObject("appVideo").getLong("play_sec");
                String userId = jsonObject.getJSONObject("common").getString("uid");
                Long ts = jsonObject.getLong("ts");

                return new VideoPlayBean(
                        "", "", "", "",
                        videoId,
                        userId,
                        0L,
                        playSec,
                        0L,
                        ts
                );
            }
        });
        SingleOutputStreamOperator<VideoPlayBean> withVideoDimDS = AsyncDataStream.unorderedWait(
                beanDS,
                new DimAsyncFunction<VideoPlayBean>("dim_video_info") {

                    @Override
                    public String getKey(VideoPlayBean videoPlayBean) {
                        return videoPlayBean.getVideoId();
                    }

                    @Override
                    public void join(VideoPlayBean videoPlayBean, JSONObject dimInfo) {
                        String chapterId = dimInfo.getString("chapter_id".toUpperCase());
                        videoPlayBean.setChapterId(chapterId);
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        SingleOutputStreamOperator<VideoPlayBean> withChapterDimDS = AsyncDataStream.unorderedWait(
                withVideoDimDS,
                new DimAsyncFunction<VideoPlayBean>("DIM_CHAPTER_INFO") {

                    @Override
                    public String getKey(VideoPlayBean videoPlayBean) {
                        return videoPlayBean.getChapterId();
                    }

                    @Override
                    public void join(VideoPlayBean videoPlayBean, JSONObject dimInfo) {
                        String chapterName = dimInfo.getString("CHAPTER_NAME");
                        videoPlayBean.setChapterName(chapterName);
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        // TODO 5. 设置水位线，以章节，视频，用户分组 ,开启会话窗口 ，聚合 ,
        //    使用键控状态，计算 播放人数
        //    使用会话窗口，计算播放次数，播放时长
        SingleOutputStreamOperator<VideoPlayBean> sessionDS = withChapterDimDS
                // 设置 事件时间水位线
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<VideoPlayBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<VideoPlayBean>() {
                                    @Override
                                    public long extractTimestamp(VideoPlayBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                })
                )
                // 分组
                .keyBy(new KeySelector<VideoPlayBean, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(VideoPlayBean value) throws Exception {
                        return Tuple4.of(value.getChapterId(), value.getChapterName(), value.getVideoId(), value.getUserId());
                    }
                })
                // 使用键控状态 计算 播放人数
                .process(new KeyedProcessFunction<Tuple4<String, String, String, String>, VideoPlayBean, VideoPlayBean>() {
                    // 保存上次观看的日期
                    private ValueState<String> lastDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> lastPlayDateVSD = new ValueStateDescriptor<>(
                                "last_play_date", String.class
                        );
                        // 设置过期时间1天
                        lastPlayDateVSD.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.days(1)).build());
                        lastDateState = getRuntimeContext().getState(lastPlayDateVSD);
                    }

                    @Override
                    public void processElement(VideoPlayBean value, Context ctx, Collector<VideoPlayBean> out) throws Exception {
                        String lastDate = lastDateState.value();
                        String curDate = DateUtil.toDate(value.getTs());
                        if (StringUtils.isEmpty(lastDate) || !curDate.equals(lastDate)) {
                            value.setPlayNum(1L);
                            lastDateState.update(curDate);
                        }
                        out.collect(value);
                    }
                })
                .keyBy(new KeySelector<VideoPlayBean, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(VideoPlayBean value) throws Exception {
                        return Tuple4.of(value.getChapterId(), value.getChapterName(), value.getVideoId(), value.getUserId());
                    }
                })
                // 开启会话窗口
                .window(EventTimeSessionWindows.withGap(Time.seconds(35)))
                .process(new ProcessWindowFunction<VideoPlayBean, VideoPlayBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple4<String, String, String, String> stringStringStringStringTuple4, Context context, Iterable<VideoPlayBean> elements, Collector<VideoPlayBean> out) throws Exception {
                        Long durSum = 0L;
                        Long playNum = 0L;
                        VideoPlayBean bean = null;
                        // 使用窗口结束时间作为 数据的水位线
                        long ts = context.window().getEnd();

                        // 聚合窗口内所有 度量值
                        for (VideoPlayBean element : elements) {
                            durSum += element.getDurSum();
                            playNum += element.getPlayNum();
                            bean = element;
                        }
                        assert bean != null;
                        out.collect(new VideoPlayBean(
                                "", "",
                                bean.getChapterId(),
                                bean.getChapterName(),
                                bean.getVideoId(),
                                bean.getUserId(),
                                1L,
                                durSum,
                                playNum,
                                ts
                        ));
                    }
                });


        // 使用 窗口状态 ， 计算 播放次数  播放时长

//        sessionDS.print(">>>>>");


        // TODO 6. 以章节分组，开窗，聚合
        SingleOutputStreamOperator<VideoPlayBean> reduceDS = sessionDS
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<VideoPlayBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<VideoPlayBean>() {
                                    @Override
                                    public long extractTimestamp(VideoPlayBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                })
                )
                .keyBy(in -> in.getChapterId())
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<VideoPlayBean>() {
                    @Override
                    public VideoPlayBean reduce(VideoPlayBean value1, VideoPlayBean value2) throws Exception {
                        value1.setPlayCt(value1.getPlayCt() + value2.getPlayCt());
                        value1.setPlayNum(value1.getPlayNum() + value2.getPlayNum());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                        return value1;
                    }
                }, new WindowFunction<VideoPlayBean, VideoPlayBean, String, TimeWindow>() {
                    @Override
                    public void apply(String keys, TimeWindow window, Iterable<VideoPlayBean> input, Collector<VideoPlayBean> out) throws Exception {
                        String stt = DateUtil.toYmdHms(window.getStart());
                        String edt = DateUtil.toYmdHms(window.getEnd());
                        for (VideoPlayBean videoPlayBean : input) {
                            videoPlayBean.setStt(stt);
                            videoPlayBean.setEdt(edt);
                            videoPlayBean.setTs(System.currentTimeMillis());
                            out.collect(videoPlayBean);
                        }
                    }
                });

        reduceDS.print(">>>>>>>>>>");


        // TODO 8. 写入clickhouse
        reduceDS.addSink(ClickhouseUtil.getSinkFunction("insert into dws_tool_video_chapter_window values(?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
